package com.atguigu.bigdata.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bigdata.edu.realtime.app.BaseAppV1;
import com.atguigu.bigdata.edu.realtime.bean.CourseExamBean;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.function.DimAsyncFunction;
import com.atguigu.bigdata.edu.realtime.util.AtguiguUtil;
import com.atguigu.bigdata.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws_14_DwsTestCourseExamWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_14_DwsTestCourseExamWindow().init(3644, 2, "Dws_14_DwsTestCourseExamWindow", Constant.TOPIC_DWD_TEST_EXAM_DETAIL);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //把流转成Javabean
        SingleOutputStreamOperator<CourseExamBean> pojo = parseToPOJO(stream);
        //开窗
        SingleOutputStreamOperator<CourseExamBean> windowAndAgg = windowAndAgg(pojo);
        //补充维度
        SingleOutputStreamOperator<CourseExamBean> addDim = addDim(windowAndAgg);

        addDim.addSink(FlinkSinkUtil.getClickHouseSink("dws_test_course_exam_window",CourseExamBean.class));
    }

    private SingleOutputStreamOperator<CourseExamBean> addDim(SingleOutputStreamOperator<CourseExamBean> windowAndAgg) {
        return AsyncDataStream.unorderedWait(windowAndAgg, new DimAsyncFunction<CourseExamBean>() {
            //从维度表读取维度，这里的课程信息维度 需要从phoenix的DIM_COURSE_INFO表读取
            @Override
            public String table() {
                return "DIM_COURSE_INFO";
            }

            @Override
            public String id(CourseExamBean input) {
                return input.getCourseId();
            }

            @Override
            public void addDim(JSONObject dim, CourseExamBean input) {
                input.setCourseName(dim.getString("COURSE_NAME"));
            }
        }, 120L, TimeUnit.SECONDS);

    }

    private SingleOutputStreamOperator<CourseExamBean> windowAndAgg(SingleOutputStreamOperator<CourseExamBean> pojo) {
        return pojo.assignTimestampsAndWatermarks(WatermarkStrategy.<CourseExamBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((o, ts) -> o.getTs())
                        .withIdleness(Duration.ofSeconds(2)))
                .keyBy(CourseExamBean::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(60)))
                .reduce(new ReduceFunction<CourseExamBean>() {
                    @Override
                    public CourseExamBean reduce(CourseExamBean value1, CourseExamBean value2) throws Exception {
                        value1.setExamPerson(value1.getExamPerson() + value2.getExamPerson());
                        value1.setAvgPoint((value1.getAvgPoint().add(value2.getAvgPoint())).divide(BigDecimal.valueOf(value1.getExamPerson())));
                        value1.setAvgTime((value1.getAvgTime().add(value2.getAvgTime())).divide(BigDecimal.valueOf(value1.getExamPerson())));
                        return value1;
                    }
                }, new ProcessWindowFunction<CourseExamBean, CourseExamBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<CourseExamBean> elements, Collector<CourseExamBean> out) throws Exception {
                        CourseExamBean next = elements.iterator().next();
                        next.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                        next.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                        next.setTs(context.currentWatermark());
                        out.collect(next);
                    }
                });
    }

    private SingleOutputStreamOperator<CourseExamBean> parseToPOJO(DataStreamSource<String> stream) {
        return stream.map(new MapFunction<String, CourseExamBean>() {
            @Override
            public CourseExamBean map(String value) throws Exception {
                JSONObject object = JSONObject.parseObject(value);
                String course_id = object.getString("course_id");
                Long score = object.getLong("score");
                Long duration_sec = object.getLong("duration_sec");
                Long ts = object.getLong("ts");
                return new CourseExamBean("", "", course_id, "", 1L, BigDecimal.valueOf(score), BigDecimal.valueOf(duration_sec), ts);
            }
        });
    }
}
